#include "../include/DbNonblock.h"
#include "../include/MsgQueue.h"

#include <list>

#define SQL_EXEC    1
#define SQL_QUERY_RESULT    2

namespace ZY
{

    list<DbNonblock::execNode* > DbNonblock::_execQueue;
    list<DbNonblock::execNode*> DbNonblock::_waitSwapExecQueue;

    pthread_mutex_t DbNonblock::queueSwapLock = PTHREAD_MUTEX_INITIALIZER;
    pthread_cond_t DbNonblock::execQueryReady = PTHREAD_COND_INITIALIZER;

    map<int, DbNonblock *> DbNonblock::_instance_map;

    DbNonblock & DbNonblock::getInstance()
    {
        return getInstance( 0 );
    }

    DbNonblock & DbNonblock::getInstance( int id )
    {
        if ( _instance_map.count( id ) == 0 ) {
            _instance_map[id] = new DbNonblock();
        } else {
            //检查是否可用

        }

        return *_instance_map[id];
    }

    DbNonblock::DbNonblock()
    {
        //ctor
    }

    DbNonblock::~DbNonblock()
    {
        //dtor
        list<execNode*>::iterator iter;

        for( iter = _waitExecQueue.begin() ;  iter != _waitExecQueue.end() ; iter ++ ) {
            delete *iter;
        }
    }

    DbNonblock & DbNonblock::QueryCallBack( SQL &sql , MsgQueue::CallbackFn _objCallBack , ByteString & bs )
    {
        string sqlStr;
        sqlStr = sql.getSQL( *this );
        return QueryCallBack( sqlStr , _objCallBack , bs );
    }

    DbNonblock & DbNonblock::QueryCallBack( SQL &sql , MsgQueue::CallbackFn _objCallBack )
    {
        string sqlStr;
        sqlStr = sql.getSQL( *this );
        return QueryCallBack( sqlStr , _objCallBack );
    }


    DbNonblock & DbNonblock::QueryCallBack( const string & sql , MsgQueue::CallbackFn _objCallBack , ByteString & bs  )
    {
        return _QueryCallBack( SQL_EXEC , sql , _objCallBack , & bs );
    }

    DbNonblock & DbNonblock::QueryCallBack( const string & sql , MsgQueue::CallbackFn _objCallBack  )
    {
        return _QueryCallBack( SQL_EXEC , sql , _objCallBack , NULL );
    }


    DbNonblock & DbNonblock::QueryResultCallBack( SQL &sql , MsgQueue::CallbackFn _objCallBack  , ByteString & bs  )
    {
        string sqlStr;
        sqlStr = sql.getSQL( *this );
        return QueryResultCallBack( sqlStr , _objCallBack , bs );
    }

    DbNonblock & DbNonblock::QueryResultCallBack( SQL &sql , MsgQueue::CallbackFn _objCallBack   )
    {
        string sqlStr;
        sqlStr = sql.getSQL( *this );
        return QueryResultCallBack( sqlStr , _objCallBack );
    }

    DbNonblock & DbNonblock::QueryResultCallBack( const string &sql , MsgQueue::CallbackFn _objCallBack  , ByteString & bs  )
    {
        return _QueryCallBack( SQL_QUERY_RESULT , sql , _objCallBack , & bs );
    }

    DbNonblock & DbNonblock::QueryResultCallBack( const string &sql , MsgQueue::CallbackFn _objCallBack   )
    {
        return _QueryCallBack( SQL_QUERY_RESULT , sql , _objCallBack , NULL );
    }


    DbNonblock & DbNonblock::_QueryCallBack( int type , const string & sql
                                            , MsgQueue::CallbackFn _objCallBack , ByteString *bs )
    {
        execNode *en;
        en = new execNode();
        en->sql.assign( sql );
        en->type = type;
        en->callback = _objCallBack;
        en->db = this;
        if ( bs ) {
            en->ext.copyFrom( *bs );
        }
        _waitExecQueue.push_back( en );
        return *this;
    }

    //将队列放入交换区
    void DbNonblock::addSwapExecQueue()
    {
        if ( pthread_mutex_trylock( & queueSwapLock ) != 0 ) {
            return;
        }

//        if ( _waitSwapExecQueue.size() == 0 ) {
//            _waitSwapExecQueue.swap( _waitExecQueue );
//        }

//        _waitSwapExecQueue.insert( _waitSwapExecQueue.end() , _waitExecQueue.begin() , _waitExecQueue.end() );
//        _waitExecQueue.clear();
        if ( _waitExecQueue.size() > 0 ) {
            _waitSwapExecQueue.splice( _waitSwapExecQueue.end() , _waitExecQueue , _waitExecQueue.begin() , _waitExecQueue.end() );
            pthread_cond_signal( & execQueryReady );
        }

        pthread_mutex_unlock( & queueSwapLock );
    }

    void DbNonblock::runExecQueueDeamon( DbNonblock & db)
    {
        pthread_t pid;
        pthread_attr_t thread_attr;

        pthread_attr_init( & thread_attr );
        pthread_attr_setdetachstate( & thread_attr ,  PTHREAD_CREATE_DETACHED ); //设置线程分离状态

        pthread_cond_init( &execQueryReady , NULL );
        pthread_mutex_init( & queueSwapLock , NULL );

        pthread_create( & pid , & thread_attr , ( void* (*) (void *) ) DbNonblock::runExecQueue , & db );
    }

    void DbNonblock::runExecQueue( DbNonblock *db )
    {
        size_t size;
        bool isLocked(false);

        while(true) {
            try {
                db->Connect();
                while( true ) {
                    pthread_mutex_lock( & queueSwapLock );
                    isLocked = true;
                    size = _waitSwapExecQueue.size();
                    while ( size == 0 ) {
                        db->Close();
                        pthread_cond_wait( & execQueryReady , & queueSwapLock );
                        db->Connect();
                        size = _waitSwapExecQueue.size(); 
                    }
                    pthread_mutex_unlock( & queueSwapLock );
                    isLocked = false;

                    _runExecQueue();
                }
            } catch ( DbException &e ) {
                cout << e.getMessage() << endl;
                if ( isLocked ) {
                    pthread_mutex_unlock( & queueSwapLock );
                }
            }
            sleep( 1 );
        }
    }

    //执行交换区的队列
    void DbNonblock::_runExecQueue( )
    {
        if ( _execQueue.size() == 0 ) {

            if ( pthread_mutex_trylock( & queueSwapLock ) != 0 ) {
                return;
            }
            if ( _waitSwapExecQueue.size() == 0 ) {
                pthread_mutex_unlock( & queueSwapLock );
                return;
            } else {
                _execQueue.swap( _waitSwapExecQueue );
            }
            pthread_mutex_unlock( & queueSwapLock );
        }
        //开始执行队列
        list< execNode* >::iterator iter;
        execNode    *en(0);
        int insertId(0) , affectRows(0);

        ByteString bs;
        DbResult *res(NULL);
        MsgQueue & msgQueue = MsgQueue::getInstance( MsgQueueType::DB_QUEUE );

        while(_execQueue.size()>0){

            en = _execQueue.front();
            try {
                bs.clear();
                bs.setInt32( en->type );
                switch( en->type ) {
                    case SQL_EXEC: //需要返回查询结果
                        en->db->Query( en->sql );
                        res = NULL;
                        insertId = en->db->GetInsertId();
                        affectRows = en->db->GetAffectedRows();
                        break;
                    case SQL_QUERY_RESULT: //需要返回结果集
                        res = en->db->QueryResult( en->sql );
                        insertId = 0;
                        affectRows = en->db->GetNumRows( res );
                        break;
                }
                //执行成功
                bs.setInt8( 1 );
            } catch ( DbException dbe ) {
                //执行失败

                bs.setInt8( 0 );
                res = NULL;
                insertId = affectRows = 0;
            }

            bs.setInt32( insertId );
            bs.setInt32( affectRows );

            bs.setUnsignedInt64( (uint64_t) res );
            string str( en->ext.getDataString() );
            bs.setString( str );

            //存入消息队列中
            msgQueue.add( en->callback , bs );

            _execQueue.pop_front();
            delete en;
        }

        //合并当前的执行队列至交换队列中
        msgQueue.merginTo( MsgQueue::getInstance( MsgQueueType::SWAP_CACHE_QUEUE ) );
    }

    DbNonblock::CallBackResult *DbNonblock::GetCallBackResult( ByteString & bs )
    {
        CallBackResult *r;
        r = new CallBackResult();
        GetCallBackResult( *r , bs );
        return r;
    }

    void DbNonblock::GetCallBackResult( CallBackResult & r , ByteString & bs )
    {
        r.type = bs.getInt32();
        r.success = bs.getInt8() == 1 ? true : false;
        r.insertId = bs.getInt32();
        r.affectRows = bs.getInt32();
        r.dbResult = ( DbResult *) bs.getUnsignedInt64();
        r.ext.copyFrom( bs );
    }

    void DbNonblock::CallBackResultFree( CallBackResult *r )
    {
        delete r;
    }

    DbNonblock::CallBackResult::~CallBackResult() {
        delete dbResult;
    }
}
